package prjCode;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.TreeSet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class CalcMi {


	public static class MapClass extends Mapper<LongWritable, Text, Text, Text> {

		private FloatWritable miX               = new FloatWritable();
		private FloatWritable miY               = new FloatWritable();
		private Text   nKey                     = new Text();
		private Text   nVal                     = new Text();
		private double nP                       = 0;
		private double nPx                      = 0;
		private double nPy                      = 0;
		private double nX                       = 0;
		private double nY                       = 0;
		private double calc                     = 0;
		private String[] arr;
		private double base                       = Math.log(2);
		private double n;

		private void calcMi(int N)
		{
			nP = Math.log(Float.parseFloat(arr[3]));
			nX = Math.log(Float.parseFloat(arr[4]));
			nY = Math.log(Float.parseFloat(arr[5]));
			nPx = Math.log(Float.parseFloat(arr[6]));
			nPy = Math.log(Float.parseFloat(arr[7]));
			n   = Math.log(N);
			
			calc= (nPx + n - nP - nX)/base;
			miX.set((float) calc);
			calc= (nPy + n - nP - nY)/base;
			miY.set((float) calc);
		}

		
		@Override
		public void map(LongWritable key, Text value, Context context) throws IOException,  InterruptedException {
			
			final Configuration conf = context.getConfiguration();

			final int N = conf.getInt("N",0);
			
			arr      = (value.toString()).split("/");
			
			if (arr.length < 8)
				return;
			
			calcMi(N);
			nKey.set(arr[0]);
			
			nVal.set(arr[1] + " " + miX + "/" + arr[2] + " " + miY );
			context.write(nKey, nVal);		
		}

	}

	public static class ReduceClass extends Reducer<Text ,Text,Text ,Text> {

		private TreeSet<String> TreeX = new TreeSet<String>();
		private TreeSet<String> TreeY = new TreeSet<String>();
		private Text retVal      = new Text();
		private double sumX = 0;
		private double sumY = 0;
		private String slotX;
		private String slotY;

		@Override
		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException,  InterruptedException {

			TreeX.clear();
			TreeY.clear();
			sumX = 0;
			sumY = 0;
			String[] slots;
			slotX = "";
			slotY = "";

			for (Text value : values) {
				slots = value.toString().split("/");
				TreeX.add(slots[0]);
				TreeY.add(slots[1]);
				sumX  += Float.parseFloat(slots[0].substring(slots[0].indexOf(" ") + 1));
				sumY  += Float.parseFloat(slots[1].substring(slots[1].indexOf(" ") + 1));
			}
			
			for(String slot : TreeX)
			{
				slotX = slotX + slot + " ";
			}
			
			for(String slot : TreeY)
			{
				slotY = slotY + slot + " ";
			}

			
			retVal.set("/" + slotX + "/" + slotY + "/" + sumX +"/" + sumY);

			context.write(key ,retVal);

		}
	}

	

	public static class PartitionerClass extends Partitioner<Text, Text> {
		// ensure that keys with same key are directed to the same reducer
		@Override
		public int getPartition(Text key,Text value, int numPartitions) {
			return  Math.abs(key.hashCode()) % numPartitions;
		}
	}

	public static void main(String[] args) throws Exception {



		Configuration conf = new Configuration();
//
//		Path p = new Path(args[2]);
//
//		String line;
//		String[] tokens;
//		BufferedReader joinReader = new BufferedReader(new FileReader(p.toString()));
//		try {
//			while ((line = joinReader.readLine()) != null) {
//				tokens = line.split("\t", 2);
//				conf.set("N", tokens[1]);
//			}
//
//		} finally {
//			joinReader.close();
//		}

		conf.set("N", "710127931");
		Job job = new Job(conf, "Calculate Mi");
		job.setJarByClass(CalcMi.class);
		job.setMapperClass(CalcMi.MapClass.class);
		job.setPartitionerClass(CalcMi.PartitionerClass.class);
		job.setReducerClass(CalcMi.ReduceClass.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.setMapOutputKeyClass(Text.class);
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}


}
